import java.sql.Timestamp
import java.text.SimpleDateFormat

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @author: yangShen
  * @Description: 统计热门页面
  * @Date: 2020/4/29 15:33 
  */

//输入数据样例类
case class ApacheLogEvent( ip: String, userId: String, eventTime: Long, method: String, url: String )

//窗口聚合结果样例类
case class UrlViewCount( url: String, windowEnd: Long, count: Long )

object NetworkFlow {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val dataStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\UserBehaviorAnalysis\\NetworkFlowAnalysis\\src\\main\\resources\\apache.log")
      .map(data => {
        val dataArray = data.split(" ")
        //定义事件转换
        val simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
        val timestamp = simpleDateFormat.parse(dataArray(3)).getTime
        ApacheLogEvent(dataArray(0), dataArray(1).trim, timestamp, dataArray(5).trim, dataArray(6).trim)
      })
      //去掉png后缀的图片
      .filter(!_.url.matches("(.*)png"))
      //指定时间戳的字段数据为乱序时，要是用如下,                                 ---设置延时时间,水位线 (1)  此处应该设置50秒，但为了效率设置为1
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(1)) {
        //设置时间戳
        override def extractTimestamp(element: ApacheLogEvent): Long = element.eventTime
      })
      .keyBy(_.url)
      //设置滑窗(10分钟一个窗口，5秒种滑动一次)
      .timeWindow(Time.minutes(10), Time.seconds(5))
      //设置延迟
      .allowedLateness(Time.seconds(60))
      .aggregate( new CountAgg(), new WindowResult())   //窗口聚合
      .keyBy(_.windowEnd)
      .process(new TopNumHotUrls(5))

    dataStream.print()

    environment.execute("network flow job")

  }
}

//自定义预聚合函数：计数器                  类型：[输入为ApacheLogEvent, 计数器, 输出累计数]
class CountAgg() extends AggregateFunction[ApacheLogEvent, Long, Long]{
  override def createAccumulator(): Long = 0L

  override def add(value: ApacheLogEvent, accumulator: Long): Long = accumulator + 1

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}
//自定义窗口函数                         参数(输入类型是第一个参数的输出, 输出类型是UrlViewCount, 窗口的key是url来源于上面的：.keyBy(_.url)操作, 窗口类型)
class WindowResult() extends  WindowFunction[Long, UrlViewCount, String, TimeWindow]{
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[UrlViewCount]): Unit = {
    out.collect( UrlViewCount(key, window.getEnd, input.iterator.next()) )
  }
}

//自定义排序输出处理函数              类型[key为windowEnd来源于前面: .keyBy(_.windowEnd), 输入类型为UrlViewCount, 输出]
class TopNumHotUrls(topSize: Int) extends KeyedProcessFunction[Long, UrlViewCount, String]{

  lazy val urlStates: ListState[UrlViewCount] = getRuntimeContext.getListState(new ListStateDescriptor[UrlViewCount]("url-state", classOf[UrlViewCount]))

  override def processElement(value: UrlViewCount, ctx: KeyedProcessFunction[Long, UrlViewCount, String]#Context, out: Collector[String]): Unit = {
    urlStates.add(value)

    //注册一个定时器
    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)
  }

  //
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, UrlViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    //从状态中拿到数据
    val allUrlViews: ListBuffer[UrlViewCount] = new ListBuffer[UrlViewCount]()
    val ite = urlStates.get().iterator()
    while (ite.hasNext){
      allUrlViews += ite.next()
    }

    urlStates.clear()

    //降序 从高到低：_.count > _.count == true
    val sortedUrlViews = allUrlViews.sortWith(_.count > _.count).take(topSize)

    //格式化结果输出
    val result = new StringBuilder()
    result.append("时间：").append(new Timestamp(timestamp -1)).append("\n")
    for (i <- sortedUrlViews.indices){
      val currentUrlView = sortedUrlViews(i)
      result.append("No").append(i + 1).append(":").append(" URL=").append(currentUrlView.url).append(" 访问量=").append(currentUrlView.count).append("\n")
    }

    result.append("=====================================")
    Thread.sleep(1000)
    out.collect(result.toString())

  }
}
